package com.atguigu.flink.datastreamapi.agg;

import org.apache.flink.api.common.functions.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created by Smexy on 2023/4/4
 *
 *  做map操作，每次map时，需要读写redis
 *      某些只需要在Task启动或关闭时只做一次的操作，可以通过实现 RichFunction 接口实现 或 继承  xxxRichFunction
 *               void open(Configuration parameters): Task被创建时，调用一次
 *               void close() : Task被关闭时，调用一次。
 *
 *
 *
 *
 *
 */
public class Demo3_RichFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        env.socketTextStream("hadoop102",8888)
           /*.map(new MapFunction<String, String>()
           {
               @Override
               public String map(String value) throws Exception {
                   System.out.println("创建连接...");
                   System.out.println("关闭连接...");
                   return value;
               }
           })*/
           .map(new MyMapFunction())
           .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    //public static class MyMapFunction implements MapFunction, RichFunction{
    public static class MyMapFunction extends RichMapFunction<String ,String>
    {

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("创建连接...");
        }

        @Override
        public void close() throws Exception {
            System.out.println("关闭连接...");
        }

        @Override
        public String map(String value) throws Exception {
            System.out.println("使用连接...");
            return value;
        }
    }
}
